package com.monkeyboy.demo.guigu.consumer;

import org.apache.kafka.clients.consumer.*;
import org.apache.kafka.common.TopicPartition;

import java.time.Duration;
import java.util.*;

public class ConsumerSample {
    private final static String TOPIC_NAME = "first";

    public static void main(String[] args) {

        helloworld();
    }
    /*
        工作里这种用法，有，但是不推荐
     */
    private static void helloworld() {
        Properties props = new Properties();
        props.setProperty("bootstrap.servers", "192.168.1.150:9092");

        //消费者组
        props.setProperty("group.id", "test");

        //自动提交消费端的offset
        props.setProperty("enable.auto.commit", "false");
        //自动提交的延时
        props.setProperty("auto.commit.interval.ms", "1000");
        //反序列化设置
        props.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

        //重置消费点,默认为lasted
        //只有消费者不同的时候才会生效，只要组没换就不会生效，或者数据过期被删除了也会导致消费不到
        //默认是lasted，所以要生效的条件就是两个，一个是换组，一个是设置值为earliest
//        props.setProperty(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,"earliest");


        KafkaConsumer<String, String> consumer = new KafkaConsumer(props);
        // 消费订阅哪一个Topic或者几个Topic
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        while (true) {
            //在这个指定时间段里批量拉取一部分可以消费的消息，时间到了就立刻返回数据。
            //由于内部是长轮询的方式，如果没有数据的话一直轮询会浪费资源，所以这里就设置一个时间表示如果没有数据就延迟100毫秒再去拉去一次返回
            ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
            //一次拉取的是多条消息
            for (ConsumerRecord<String, String> record : records)
                System.out.printf("patition = %d , offset = %d, key = %s, value = %s%n",
                        record.partition(), record.offset(), record.key(), record.value());
            consumer.commitAsync();//手动提交offset
        }
    }

}
